package cn.virens.server.tio_tcp;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantReadWriteLock.ReadLock;
import java.util.function.Consumer;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.tio.core.ChannelContext;
import org.tio.core.PacketHandlerMode;
import org.tio.core.Tio;
import org.tio.server.ServerTioConfig;
import org.tio.server.TioServer;
import org.tio.utils.lock.SetWithLock;
import org.tio.utils.thread.pool.DefaultThreadFactory;
import org.tio.utils.thread.pool.SynThreadPoolExecutor;

import cn.virens.app.AppConfig;
import cn.virens.app.cmpt.Component;
import cn.virens.server.tio_tcp.handler.PacketHandler;

public class TIOServerManagerImpl implements TIOServerManager, Component {
	private final Logger logger = LoggerFactory.getLogger(TIOServerManagerImpl.class);

	private final List<PacketHandler> handlerPool = new ArrayList<>();

	private final TioServer mTioServer = new TioServer(null);

	private static final long KEEP_ALIVE_TIME = 90L;
	private static final int CORE_POOL_SIZE = Runtime.getRuntime().availableProcessors() * 2;
	private static final int MAX_POOL_SIZE = CORE_POOL_SIZE * 4 < 128 ? 128 : CORE_POOL_SIZE * 4;

	private ThreadPoolExecutor groupExecutor;
	private SynThreadPoolExecutor tioExecutor;

	private ServerTioConfig serverTioConfig;
	private TIOServerHandler serverTioHandler;
	private TIOServerListener serverTioListener;

	/** 添加 数据包解析器 */
	public final void addHandler(PacketHandler handler) {
		synchronized (handlerPool) {
			this.handlerPool.add(handler);
		}
	}

	@Override
	public ChannelContext channelGet(String mac) {
		return Tio.getByBsId(serverTioConfig, mac);
	}

	@Override
	public void channelForEach(Consumer<ChannelContext> consumer) {
		SetWithLock<ChannelContext> setWithLock = this.serverTioConfig.connections;
		ReadLock readLock = setWithLock.readLock();

		readLock.lock();
		try {
			for (ChannelContext channelContext : setWithLock.getObj()) {
				consumer.accept(channelContext);
			}
		} catch (Throwable e) {
			logger.error(e.getMessage(), e);
		} finally {
			readLock.unlock();
		}
	}

	@Override
	public void initialize(AppConfig setting) throws Exception {
		// TIO 执行队列
		LinkedBlockingQueue<Runnable> tioQueue = new LinkedBlockingQueue<>();
		DefaultThreadFactory threadFactory0 = DefaultThreadFactory.getInstance("tio-tcp", Thread.NORM_PRIORITY);
		this.tioExecutor = new SynThreadPoolExecutor(CORE_POOL_SIZE, CORE_POOL_SIZE, KEEP_ALIVE_TIME, tioQueue, threadFactory0, "tio-tcp");
		this.tioExecutor.prestartAllCoreThreads();

		// TIO 分组队列
		LinkedBlockingQueue<Runnable> groupQueue = new LinkedBlockingQueue<>();
		DefaultThreadFactory threadFactory1 = DefaultThreadFactory.getInstance("tio-tcp-group", Thread.NORM_PRIORITY);
		this.groupExecutor = new ThreadPoolExecutor(MAX_POOL_SIZE, MAX_POOL_SIZE, KEEP_ALIVE_TIME, TimeUnit.SECONDS, groupQueue, threadFactory1);
		this.groupExecutor.prestartAllCoreThreads();

		// 初始化TIO
		this.serverTioListener = new TIOServerListener(handlerPool);
		this.serverTioHandler = new TIOServerHandler(handlerPool);

		this.serverTioConfig = new ServerTioConfig(serverTioHandler, serverTioListener, tioExecutor, groupExecutor);
		this.serverTioConfig.setHeartbeatTimeout(setting.getLong("tio.tcp.heartbeat"));
		this.serverTioConfig.setPacketHandlerMode(PacketHandlerMode.QUEUE);
		this.serverTioConfig.setName("TCP ServerGroup");
		this.serverTioConfig.statOn = false;

		// 启动TCP服务
		this.mTioServer.setServerTioConfig(serverTioConfig);
		this.mTioServer.start(null, setting.getInteger("tio.tcp.port"));
	}

	@Override
	public void destroy() throws Exception {
		this.mTioServer.stop();
	}
}
